package cn.edu.thu.laud.auth;
/*
 * 
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * 
 */


import java.nio.ByteBuffer;
import java.util.EnumSet;
import java.util.List;

import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.IAuthority;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.auth.Resources;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.thrift.LaUDAdministrator;

public class SimpleAuthority extends MemoryAuthenticator implements IAuthority
{
    public final static String ACCESS_FILENAME_PROPERTY = "access.properties";
    // magical property for WRITE permissions to the keyspaces list
    public final static String KEYSPACES_WRITE_PROPERTY = "<modify-keyspaces>";
    Logger logger= LoggerFactory.getLogger(SimpleAuthority.class);
    
    

    
    
    public EnumSet<Permission> authorize(AuthenticatedUser user, List<Object> resource)
    {
    	logger.debug("authorize.user:"+user+"--begin");
    	logger.debug(Resources.toString(resource));
    	logger.debug("----");
        if (resource.size() < 2 || !Resources.ROOT.equals(resource.get(0)) || !Resources.KEYSPACES.equals(resource.get(1)))
            return Permission.NONE;
        //daemonClient can do everything..
        if(user.username.equals("root")){
        	return Permission.ALL;
        }
        if(user.username.equals("testadmin")){
        	return Permission.ALL;
        }
        String keyspace, columnFamily = null;
        EnumSet<Permission> authorized = Permission.NONE;
        
        // /cassandra/keyspaces
        if (resource.size() == 2)
        {
            keyspace = KEYSPACES_WRITE_PROPERTY;
            authorized = EnumSet.of(Permission.READ);
        }
        // /cassandra/keyspaces/<keyspace name>
        else if (resource.size() == 3)
        {
            keyspace = (String)resource.get(2);
        }
        // /cassandra/keyspaces/<keyspace name>/<cf name>
        else if (resource.size() == 4)
        {
            keyspace = (String)resource.get(2);
            columnFamily = (String)resource.get(3);
        }
        else
        {
            // We don't currently descend any lower in the hierarchy.
            throw new UnsupportedOperationException();
        }
        List<ColumnOrSuperColumn> groups=userGroupAuthorities.get(user.username);
        List<ColumnOrSuperColumn> ksAuthorities=userKsAuthorities.get(user.username);
        
//        String accessFilename = System.getProperty(ACCESS_FILENAME_PROPERTY);
//        InputStream in=null;
//        try
//        {
//            in = new BufferedInputStream(new FileInputStream(accessFilename));
//            Properties accessProperties = new Properties();
//            accessProperties.load(in);
//
//            
            // Special case access to the keyspace list
            if (keyspace == KEYSPACES_WRITE_PROPERTY||columnFamily == null)
            {
            	if(groups!=null){
            		for(ColumnOrSuperColumn columnOrSuperColumn:groups){
            			if(columnOrSuperColumn.column.name.equals(LaUDAdministrator.LAUD_SYSTEM_COL_GROUP_BYTE)
            					&&Int32Type.instance.compose(columnOrSuperColumn.column.value)>=AuthorityGroup.DBDESIGN.getValue()
            			   ){
            				return Permission.ALL;
            			}
            		}
            		return  EnumSet.of(Permission.READ);
            	}else{
            		return  EnumSet.of(Permission.READ);
            	}
            }

            boolean canRead = false, canWrite = false;
            
//            if (columnFamily == null)
//            {    
//                //readers = accessProperties.getProperty(keyspace + ".<ro>");
//               // writers = accessProperties.getProperty(keyspace + ".<rw>");
//            	logger.debug("cf is null!!!!!");
//            	return EnumSet.of(Permission.READ);
//            }
           
            
            ByteBuffer keyspaceByteBuffer=UTF8Type.instance.decompose(keyspace);
            ByteBuffer columnFamilyByteBuffer=UTF8Type.instance.decompose(columnFamily);
            int authority=0;
            if(ksAuthorities!=null){
            	for(ColumnOrSuperColumn columnOrSuperColumn:ksAuthorities){
            		if(columnOrSuperColumn.super_column.name.equals(keyspace)){
            			for(Column column:columnOrSuperColumn.super_column.columns){
            				if(column.name.equals(columnFamilyByteBuffer)){
            					authority=Int32Type.instance.compose(column.value);
            				}
            			}
            		}
            	}
            }
            if(authority/2==1){
            	canWrite=true;
            	authority%=2;
            }
            if(authority==1){
            	canRead=true;
            }
            
            if (canWrite)
                authorized = Permission.ALL;
            else if (canRead)
                authorized = EnumSet.of(Permission.READ);

            return authorized;
    }

    public void validateConfiguration() throws ConfigurationException
    {
    	logger.debug("validateConfiguration");
//        String afilename = System.getProperty(ACCESS_FILENAME_PROPERTY);
//        if (afilename == null)
//        {
//            throw new ConfigurationException(String.format("When using %s, '%s' property must be defined.",
//                                                           this.getClass().getCanonicalName(),
//                                                           ACCESS_FILENAME_PROPERTY));	
//        }
    }
}
